package com.ict.utils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;

import org.apache.log4j.Logger;

import com.ict.config.Config;
import com.ict.dataprocess.FuseMQToRedis;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

import team.iOceanPlus.PB.AISData.PBAISDynamic;
import team.iOceanPlus.PB.Target;

/**
 * @author TLC
 * 连接MQ工具类
 *
 */

public class ConnMQUtil {
	private static  Logger  logger=Logger.getLogger(ConnMQUtil.class);
	public static Channel sendChannel;
    public static Connection sendConnection;
    public static Channel recvChannel;
    public static Connection recvConnection;
    public static QueueingConsumer consumer;
    
    /**
     * 生产者连接初始化
     */
    static {
        ConnectionFactory sendFactory = new ConnectionFactory();
        sendFactory.setHost(Config.sendMQIP);
        sendFactory.setUsername(Config.sendMQUserName);
        sendFactory.setPassword(Config.sendMQPasswd);
        sendFactory.setPort(Config.sendMQPort);
       //sendFactory.setVirtualHost(Config.SendVirtualHost);
        
        try {
			sendConnection = sendFactory.newConnection();
			sendChannel=sendConnection.createChannel();
			logger.info("connect  send mq success! ");
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (TimeoutException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    	
    }
    /**
     * 消费者连接初始化
     */
    static {
        ConnectionFactory recvFactory = new ConnectionFactory();
        recvFactory.setHost(Config.recvMQIP);
        recvFactory.setUsername(Config.recvMQUserName);
        recvFactory.setPassword(Config.recvMQPasswd);
        recvFactory.setPort(Config.recvMQPort);
      // recvFactory.setVirtualHost(Config.recvVirtualHost);
        recvFactory.setAutomaticRecoveryEnabled(true);
        try {
			recvConnection = recvFactory.newConnection();
			recvChannel=recvConnection.createChannel();
			consumer = new QueueingConsumer(recvChannel);
			logger.info("connect  recv mq success! ");
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (TimeoutException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    	
    }
    

    /**
     * 
     * 生产者关闭通道、连接
     */
    public void sendClose() throws IOException{
        try{
            sendChannel.close();
        } catch (TimeoutException ex){
            System.out.println("ex" + ex);
        }
        sendConnection.close();
    }
    /**
     * 
     * 消费者关闭通道、连接
     */
    public void recvClose() throws IOException{
        try{
            recvChannel.close();
        } catch (TimeoutException ex){
            System.out.println("ex" + ex);
        }
        recvConnection.close();
    }
    
    public  void send(String exchangeType,String exchangeName,String rootingKey,String queueName,byte [] message) throws IOException {
    	if(exchangeType.equals("fanout")) {
    		try {
    			sendChannel.exchangeDeclare(exchangeName,"fanout");
				sendChannel.basicPublish(exchangeName, "", null, message);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
    	}
    	if(exchangeType.equals("direct")) {
    		try {
    			sendChannel.exchangeDeclare(exchangeName,"direct",true);
				sendChannel.basicPublish(exchangeName,rootingKey,null,message);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} 
    	}
    	if(exchangeType.equals("topic")) {
    		 try {
    			sendChannel.exchangeDeclare(exchangeName,"topic",true);
				sendChannel.basicPublish(exchangeName,rootingKey,null,message);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
    	}
    }
    
    
    
    
    public  QueueingConsumer.Delivery recv(String exchangeType,String exchangeName,String [] rootingKey,String queueName) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException{
    	/*List<String>  lstRecv=new ArrayList<String>();*/
    	QueueingConsumer.Delivery delivery=null;
    	if(exchangeType.equals("fanout")) {
    		recvChannel.queueDeclare(queueName,false,false,false,null);
            //绑定队列到交换器
    		recvChannel.queueBind(queueName,exchangeName,""); //不设置路由键
            //统一时刻服务器只会发一条消息给消费者;
    		//recvChannel.basicQos(1);
            //定义队列的消费者
            consumer = new QueueingConsumer(recvChannel);
            //监听队列，手动返回完成
            recvChannel.basicConsume(queueName,true,consumer);
            //获取消息
           /* while (true)
            {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(message);
                lstRecv.add(message);
                *//**确认消息*//*
                recvChannel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            }*/

    	}
    	if(exchangeType.equals("direct")) {
    		    recvChannel.queueDeclare(queueName,false,false,false,null);
    	        //绑定队列到交换机;
    		    for (String strRootKey :rootingKey) {
    		    	recvChannel.queueBind(queueName,exchangeName,strRootKey);  
    		    }
    	        //同一时刻服务器只会发送一条消息给消费者;
    		    recvChannel.basicQos(1);
    	        consumer = new QueueingConsumer(recvChannel);
    	        //监听队列，手动返回完成
    	        recvChannel.basicConsume(queueName,false,consumer);
    	        //获取消息
    	        /*while (true)
    	        {
    	            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    	            String message = new String(delivery.getBody());
    	            lstRecv.add(message);
    	            *//**确认消息*//*
    	            recvChannel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
    	        }*/

    	}
    	if(exchangeType.equals("topic")) {
    		   recvChannel.queueDeclare(queueName,false,false,false,null);
               //绑定exchange
    		   for (String strRootKey :rootingKey) {
   		    	recvChannel.queueBind(queueName,exchangeName,strRootKey);  
   		       }
               //同一时刻服务器只能发送一条消息给消费者；
    		   //recvChannel.basicQos(1);
               //声明消费者
              
               //监控队列，设置手动完成
               recvChannel.basicConsume(queueName,true,consumer);
              /* while (true)
               {
            	   QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            	   String message = new String(delivery.getBody());
            	   Target.PBTarget info=Target.PBTarget.parseFrom(delivery.getBody());
            	   for(int x=0 ;x<info.getListTargetPosList().size(); x++) {
            		   Target.PBTargetPosition  pos=info.getListTargetPosList().get(x);
            		   System.out.println(pos.getCountryName());
            		   PBAISDynamic aisDymc= pos.getAisDynamic();
            		   System.out.println(aisDymc.getIntLatitudeX60W());
            		   System.out.println(aisDymc.getIntLongitudeX60W());
            	   }
   	               *//**确认消息*//*
                   recvChannel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
                }*/

    	}
    	delivery = consumer.nextDelivery();
    	return delivery;
    }
    


}
